package context

import (
	"context"
	"errors"
	"github.com/Unixeno/TheMoon/logging"
	"github.com/Unixeno/TheMoon/model"
	"github.com/Unixeno/TheMoon/moon/starlink"
	"time"
)

var (
	ShutdownError  = errors.New("context closed")
	NotExistsError = errors.New("channel not exist")
)

var localContext *messageContext
var localStreamContext *streamContext

var logger = logging.GetLogger("context")

func InitContext() {
	ctx, cancel := context.WithCancel(context.Background())
	localContext = &messageContext{
		ctx:    ctx,
		cancel: cancel,
	}
	localStreamContext = &streamContext{}
}

func RegisterClient(client starlink.Client) {
	localContext.client = client
	errChan := client.Init(localContext.ctx)

	go func() {
		err := <-errChan
		if err != nil {
			logger.WithError(err).Error("client error, shutdown...")
			Cancel()
		}
	}()

	go recv()
}

func RegisterModule(module string) {
	localContext.registerModule(module)
}

func RegisterStream(module string, operation int, callback StreamCallback) {
	logger.WithField("module", module).WithField("op", operation).Debug("register stream callback")
	localStreamContext.registerCallback(module, operation, callback)
}

func NewStream(src string, module string, operation int) model.Stream {
	stream := newMoonStream(src, module, operation, localContext.client)
	localStreamContext.registerStream(stream.ID(), stream)
	return stream
}

func Cancel() {
	localContext.cancel()
}

func Done() <-chan struct{} {
	return localContext.ctx.Done()
}

func NewContext() (context.Context, context.CancelFunc) {
	return context.WithCancel(localContext.ctx)
}

// 异步发送消息
func SendMessage(msg model.Message) error {
	return localContext.client.Send(&msg)
}

func RecvMessage(modelName string) (model.Message, error) {
	if channel := localContext.getChan(modelName); channel != nil {
		if message, ok := <-channel; ok {
			return message, nil
		}
		return model.Message{}, ShutdownError
	}
	return model.Message{}, NotExistsError
}

func recv() {
	logger.Debug("start recv from client")
	for {
		message, err := localContext.client.Receive()
		if err != nil {
			logger.WithError(err).Error("failed to receive message")
			localContext.cancel()
			return
		}

		if message.Header.Stream {
			// 处理流消息
			handleStream(message)
			continue
		}

		// 只有parent id不为空的时候才说明是响应消息
		if message.Header.Sync && message.Header.ParentID != "" {
			// 同步消息有专门的channel
			ch := localContext.getSyncChan(message.Header.ParentID)
			if ch == nil {
				logger.WithField("id", message.Header.ParentID).Warning("invalid sync message parent id")
			} else {
				ch <- *message
			}
		} else {
			ch, ok := localContext.channels.Load(message.Header.Destination)
			if !ok {
				logger.WithField("destination", message.Header.Destination).Warning("unknown module destination")
			} else {
				ch.(chan model.Message) <- *message
			}
		}
	}
}

func handleStream(message *model.Message) {
	content := message.Content.([]byte)
	// 首先检查流是否存在
	stream := localStreamContext.getStream(message.Header.ID)
	if stream == nil {
		stream = getMoonStream(message, localContext.client)
		stream.Header.ID = message.Header.ID
		// 流不存在，接下来判断是否是握手包
		if !message.Header.Sync {
			if content == nil {
				// 忽略掉关闭链接message
				return
			}
			// 检查是否注册回调
			callback := localStreamContext.getCallback(message.Header.Destination, int(message.Operation))
			if nil == callback {
				logger.WithField("module", message.Header.Destination).
					WithField("op", message.Operation).Error("stream callback unknown")
				// 找不到回调函数，关闭流
				stream.cleanup()
				return
			} else {
				// 启动回调函数
				logger.Info("stream handshake success")
				localStreamContext.registerStream(stream.ID(), stream)
				go func() {
					callback(stream)
					stream.cleanup()
				}()
				stream.write(content)
			}
		} else {
			// 不是握手包则直接关闭流
			stream.cleanup()
		}
	} else {
		// 流存在，判断消息类型
		if message.Header.Sync {
			if content == nil {
				logger.WithField("id", stream.ID()).Debug("stream heartbeat")
				stream.Send(nil)
			} else {
				// 普通消息
				stream.write(content)
			}
		} else {
			// 关闭链接请求
			logger.WithField("id", stream.ID()).Debug("stream closed by remote")
			stream.cleanup()
		}
	}
}

// 同步消息
func SyncSendMessage(msg model.Message, timeout time.Duration) (model.Message, error) {
	logger.Debug("send sync")
	msg.Header.Sync = true
	ch := localContext.registerSyncChan(msg.Header.ID)

	defer localContext.unregisterSyncChan(msg.Header.ID)

	ctx, _ := context.WithTimeout(localContext.ctx, timeout)

	if err := localContext.client.Send(&msg); err != nil {
		return model.Message{}, err
	}
	select {
	case <-ctx.Done():
		return model.Message{}, ctx.Err()
	case msg := <-ch:
		return msg, nil
	}
}
